package com.millstein.realtime.util;

import com.millstein.realtime.common.Constants;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;
import java.util.Properties;

public class FlinkSourceUtil {

    /**
     * 获取kafka source
     * @param appName 应用名称
     * @param topicName 主题
     * @return
     */
    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String appName, String topicName) {
        Properties props = new Properties();
        // 设置kafka服务器地址
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_SERVER);
        // 设置消费组名称
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, appName);
        // 设置隔离级别。现在消费的数据只是kafka提交的数据，不提交的数据不能消费
        props.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        return new FlinkKafkaConsumer<>(
                topicName,
                // 泛型是输入数据的类型
                new KafkaDeserializationSchema<String>() {
                    /**
                     * 遇到什么样的数据要结束对流中的数据进行反序列化
                     * @param nextElement 数据
                     * @return
                     */
                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    /**
                     * 反序列化的逻辑
                     * @param record 待反序列化的数据
                     * @return
                     * @throws Exception
                     */
                    @Override
                    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                        byte[] value = record.value();
                        if (value == null) {
                            return null;
                        }
                        return new String(value, StandardCharsets.UTF_8);
                    }

                    /**
                     * 反序列化后的数据类型
                     * @return
                     */
                    @Override
                    public TypeInformation<String> getProducedType() {
                        return TypeInformation.of(new TypeHint<String>() {});
                    }
                }, props
        );
    }
}
